{
  "cells": [
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ur8xi4C7S06n"
      },
      "outputs": [],
      "source": [
        "# Copyright 2021 Google LLC\n",
        "#\n",
        "# Licensed under the Apache License, Version 2.0 (the \"License\");\n",
        "# you may not use this file except in compliance with the License.\n",
        "# You may obtain a copy of the License at\n",
        "#\n",
        "#     https://www.apache.org/licenses/LICENSE-2.0\n",
        "#\n",
        "# Unless required by applicable law or agreed to in writing, software\n",
        "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
        "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
        "# See the License for the specific language governing permissions and\n",
        "# limitations under the License."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "JAPoU8Sm5E6e"
      },
      "source": [
        "<table align=\"left\">\n",
        "\n",
        "  <td>\n",
        "    <a href=\"https://colab.research.google.com/github/GoogleCloudPlatform/ai-platform-samples/blob/master/ai-platform-unified/notebooks/official/pipelines/pipelines_intro_kfp.ipynb\">\n",
        "      <img src=\"https://cloud.google.com/ml-engine/images/colab-logo-32px.png\" alt=\"Colab logo\"> Run in Colab\n",
        "    </a>\n",
        "  </td>\n",
        "  <td>\n",
        "    <a href=\"https://github.com/GoogleCloudPlatform/ai-platform-samples/blob/master/ai-platform-unified/notebooks/official/pipelines/pipelines_intro_kfp.ipynb\">\n",
        "      <img src=\"https://cloud.google.com/ml-engine/images/github-logo-32px.png\" alt=\"GitHub logo\">\n",
        "      View on GitHub\n",
        "    </a>\n",
        "  </td>\n",
        "  <td>\n",
        "    <a href=\"https://console.cloud.google.com/ai/platform/notebooks/deploy-notebook?download_url=https://github.com/GoogleCloudPlatform/ai-platform-samples/raw/master/ai-platform-unified/notebooks/official/pipelines/pipelines_intro_kfp.ipynb\">\n",
        "      Open in Google Cloud Notebooks\n",
        "    </a>\n",
        "  </td>\n",
        "</table>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "o2rM_9Ml7-W2"
      },
      "source": [
        "# Introduction to Vertex Pipelines using the KFP SDK"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "tvgnzT1CKxrO"
      },
      "source": [
        "## Overview\n",
        "\n",
        "This notebook provides an introduction to using [Vertex Pipelines](https://cloud.google.com/vertex-ai/docs/pipelines) with [the Kubeflow Pipelines (KFP) SDK](https://www.kubeflow.org/docs/components/pipelines/).\n",
        "\n",
        "### Objective\n",
        "\n",
        "In this example, you learn:\n",
        "\n",
        "- How to define and compile a pipeline.\n",
        "- How to schedule recurring pipeline runs.\n",
        "- How to specify which service account to use for a pipeline run.\n",
        "\n",
        "\n",
        "### Costs \n",
        "\n",
        "This tutorial uses billable components of Google Cloud:\n",
        "\n",
        "* Vertex AI Training\n",
        "* Cloud Storage\n",
        "* Cloud Functions\n",
        "* Cloud Scheduler\n",
        "\n",
        "\n",
        "Learn about pricing for [Vertex AI](https://cloud.google.com/vertex-ai/pricing), [Cloud Storage](https://cloud.google.com/storage/pricing), [Cloud Functions](https://cloud.google.com/functions/pricing), and [Cloud Scheduler](https://cloud.google.com/scheduler/pricing), and use the [Pricing\n",
        "Calculator](https://cloud.google.com/products/calculator/)\n",
        "to generate a cost estimate based on your projected usage."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ze4-nDLfK4pw"
      },
      "source": [
        "### Set up your local development environment\n",
        "\n",
        "**If you are using Colab or Google Cloud Notebooks**, your environment already meets\n",
        "all the requirements to run this notebook. You can skip this step."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "gCuSR8GkAgzl"
      },
      "source": [
        "**Otherwise**, make sure your environment meets this notebook's requirements.\n",
        "You need the following:\n",
        "\n",
        "* The Google Cloud SDK\n",
        "* Git\n",
        "* Python 3\n",
        "* virtualenv\n",
        "* Jupyter notebook running in a virtual environment with Python 3\n",
        "\n",
        "The Google Cloud guide to [Setting up a Python development\n",
        "environment](https://cloud.google.com/python/setup) and the [Jupyter\n",
        "installation guide](https://jupyter.org/install) provide detailed instructions\n",
        "for meeting these requirements. The following steps provide a condensed set of\n",
        "instructions:\n",
        "\n",
        "1. [Install and initialize the Cloud SDK.](https://cloud.google.com/sdk/docs/)\n",
        "\n",
        "1. [Install Python 3.](https://cloud.google.com/python/setup#installing_python)\n",
        "\n",
        "1. [Install\n",
        "   virtualenv](https://cloud.google.com/python/setup#installing_and_using_virtualenv)\n",
        "   and create a virtual environment that uses Python 3. Activate the virtual environment.\n",
        "\n",
        "1. To install Jupyter, run `pip install jupyter` on the\n",
        "command-line in a terminal shell.\n",
        "\n",
        "1. To launch Jupyter, run `jupyter notebook` on the command-line in a terminal shell.\n",
        "\n",
        "1. Open this notebook in the Jupyter Notebook Dashboard."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "i7EUnXsZhAGF"
      },
      "source": [
        "### Install additional packages\n",
        "\n",
        "Install the KFP SDK."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "IaYsrh0Tc17L"
      },
      "outputs": [],
      "source": [
        "import os\n",
        "\n",
        "# The Google Cloud Notebook product has specific requirements\n",
        "IS_GOOGLE_CLOUD_NOTEBOOK = os.path.exists(\"/opt/deeplearning/metadata/env_version\")\n",
        "\n",
        "# Google Cloud Notebook requires dependencies to be installed with '--user'\n",
        "USER_FLAG = \"\"\n",
        "if IS_GOOGLE_CLOUD_NOTEBOOK:\n",
        "    USER_FLAG = \"--user\""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "aR7LNYMUCVKc"
      },
      "outputs": [],
      "source": [
        "! pip install {USER_FLAG} kfp --upgrade"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "hhq5zEbGg0XX"
      },
      "source": [
        "### Restart the kernel\n",
        "\n",
        "After you install the additional packages, you need to restart the notebook kernel so it can find the packages."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "EzrelQZ22IZj"
      },
      "outputs": [],
      "source": [
        "# Automatically restart kernel after installs\n",
        "import os\n",
        "\n",
        "if not os.getenv(\"IS_TESTING\"):\n",
        "    # Automatically restart kernel after installs\n",
        "    import IPython\n",
        "\n",
        "    app = IPython.Application.instance()\n",
        "    app.kernel.do_shutdown(True)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "6GPgNN7eeX1l"
      },
      "source": [
        "Check the version of the package you installed.  The KFP SDK version should be >=1.6."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "NN0mULkEeb84"
      },
      "outputs": [],
      "source": [
        "!python3 -c \"import kfp; print('KFP SDK version: {}'.format(kfp.__version__))\""
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "lWEdiXsJg0XY"
      },
      "source": [
        "## Before you begin\n",
        "\n",
        "This notebook does not require a GPU runtime."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "BF1j6f9HApxa"
      },
      "source": [
        "### Set up your Google Cloud project\n",
        "\n",
        "**The following steps are required, regardless of your notebook environment.**\n",
        "\n",
        "1. [Select or create a Google Cloud project](https://console.cloud.google.com/cloud-resource-manager). When you first create an account, you get a $300 free credit towards your compute/storage costs.\n",
        "\n",
        "1. [Make sure that billing is enabled for your project](https://cloud.google.com/billing/docs/how-to/modify-project).\n",
        "\n",
        "1. [Enable the Vertex AI, Cloud Storage, and Compute Engine APIs](https://console.cloud.google.com/flows/enableapi?apiid=aiplatform.googleapis.com,compute_component,storage-component.googleapis.com). \n",
        "\n",
        "1. Follow the \"**Configuring your project**\" instructions from the Vertex Pipelines documentation.\n",
        "\n",
        "1. If you are running this notebook locally, you will need to install the [Cloud SDK](https://cloud.google.com/sdk).\n",
        "\n",
        "1. Enter your project ID in the cell below. Then run the cell to make sure the\n",
        "Cloud SDK uses the right project for all the commands in this notebook.\n",
        "\n",
        "**Note**: Jupyter runs lines prefixed with `!` as shell commands, and it interpolates Python variables prefixed with `$` into these commands."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "WReHDGG5g0XY"
      },
      "source": [
        "#### Set your project ID\n",
        "\n",
        "**If you don't know your project ID**, you may be able to get your project ID using `gcloud`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "oM1iC_MfAts1"
      },
      "outputs": [],
      "source": [
        "import os\n",
        "\n",
        "PROJECT_ID = \"\"\n",
        "\n",
        "# Get your Google Cloud project ID from gcloud\n",
        "if not os.getenv(\"IS_TESTING\"):\n",
        "    shell_output=!gcloud config list --format 'value(core.project)' 2>/dev/null\n",
        "    PROJECT_ID = shell_output[0]\n",
        "    print(\"Project ID: \", PROJECT_ID)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "qJYoRfYng0XZ"
      },
      "source": [
        "Otherwise, set your project ID here."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "riG_qUokg0XZ"
      },
      "outputs": [],
      "source": [
        "if PROJECT_ID == \"\" or PROJECT_ID is None:\n",
        "    PROJECT_ID = \"python-docs-samples-tests\"  # @param {type:\"string\"}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "06571eb4063b"
      },
      "source": [
        "#### Timestamp\n",
        "\n",
        "If you are in a live tutorial session, you might be using a shared test account or project. To avoid name collisions between users on resources created, you create a timestamp for each instance session, and append it onto the name of resources you create in this tutorial."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "697568e92bd6"
      },
      "outputs": [],
      "source": [
        "from datetime import datetime\n",
        "\n",
        "TIMESTAMP = datetime.now().strftime(\"%Y%m%d%H%M%S\")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "dr--iN2kAylZ"
      },
      "source": [
        "### Authenticate your Google Cloud account\n",
        "\n",
        "**If you are using Google Cloud Notebooks**, your environment is already\n",
        "authenticated. Skip this step."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "sBCra4QMA2wR"
      },
      "source": [
        "**If you are using Colab**, run the cell below and follow the instructions\n",
        "when prompted to authenticate your account via oAuth.\n",
        "\n",
        "**Otherwise**, follow these steps:\n",
        "\n",
        "1. In the Cloud Console, go to the [**Create service account key**\n",
        "   page](https://console.cloud.google.com/apis/credentials/serviceaccountkey).\n",
        "\n",
        "2. Click **Create service account**.\n",
        "\n",
        "3. In the **Service account name** field, enter a name, and\n",
        "   click **Create**.\n",
        "\n",
        "4. In the **Grant this service account access to project** section, click the **Role** drop-down list. Type \"Vertex AI\"\n",
        "into the filter box, and select\n",
        "   **Vertex AI Administrator**. Type \"Storage Object Admin\" into the filter box, and select **Storage Object Admin**.\n",
        "\n",
        "5. Click *Create*. A JSON file that contains your key downloads to your\n",
        "local environment.\n",
        "\n",
        "6. Enter the path to your service account key as the\n",
        "`GOOGLE_APPLICATION_CREDENTIALS` variable in the cell below and run the cell."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "PyQmSRbKA8r-"
      },
      "outputs": [],
      "source": [
        "import os\n",
        "import sys\n",
        "\n",
        "# If you are running this notebook in Colab, run this cell and follow the\n",
        "# instructions to authenticate your GCP account. This provides access to your\n",
        "# Cloud Storage bucket and lets you submit training jobs and prediction\n",
        "# requests.\n",
        "\n",
        "# The Google Cloud Notebook product has specific requirements\n",
        "IS_GOOGLE_CLOUD_NOTEBOOK = os.path.exists(\"/opt/deeplearning/metadata/env_version\")\n",
        "\n",
        "# If on Google Cloud Notebooks, then don't execute this code\n",
        "if not IS_GOOGLE_CLOUD_NOTEBOOK:\n",
        "    if \"google.colab\" in sys.modules:\n",
        "        from google.colab import auth as google_auth\n",
        "\n",
        "        google_auth.authenticate_user()\n",
        "\n",
        "    # If you are running this notebook locally, replace the string below with the\n",
        "    # path to your service account key and run this cell to authenticate your GCP\n",
        "    # account.\n",
        "    elif not os.getenv(\"IS_TESTING\"):\n",
        "        %env GOOGLE_APPLICATION_CREDENTIALS ''"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "NxhCPW6e46EF"
      },
      "source": [
        "### Create a Cloud Storage bucket as necessary\n",
        "\n",
        "You will need a Cloud Storage bucket for this example.  If you don't have one that you want to use, you can make one now.\n",
        "\n",
        "\n",
        "Set the name of your Cloud Storage bucket below. It must be unique across all\n",
        "Cloud Storage buckets.\n",
        "\n",
        "You may also change the `REGION` variable, which is used for operations\n",
        "throughout the rest of this notebook. Make sure to [choose a region where Vertex AI services are\n",
        "available](https://cloud.google.com/vertex-ai/docs/general/locations#available_regions). You may\n",
        "not use a Multi-Regional Storage bucket for training with Vertex AI."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "MzGDU7TWdts_"
      },
      "outputs": [],
      "source": [
        "BUCKET_NAME = \"gs://[your-bucket-name]\"  # @param {type:\"string\"}\n",
        "REGION = \"us-central1\"  # @param {type:\"string\"}"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "cf221059d072"
      },
      "outputs": [],
      "source": [
        "if BUCKET_NAME == \"\" or BUCKET_NAME is None or BUCKET_NAME == \"gs://[your-bucket-name]\":\n",
        "    BUCKET_NAME = \"gs://\" + PROJECT_ID + \"aip-\" + TIMESTAMP"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-EcIXiGsCePi"
      },
      "source": [
        "**Only if your bucket doesn't already exist**: Run the following cell to create your Cloud Storage bucket."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "NIq7R4HZCfIc"
      },
      "outputs": [],
      "source": [
        "! gsutil mb -l $REGION $BUCKET_NAME"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ucvCsknMCims"
      },
      "source": [
        "Finally, validate access to your Cloud Storage bucket by examining its contents:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "vhOb7YnwClBb"
      },
      "outputs": [],
      "source": [
        "! gsutil ls -al $BUCKET_NAME"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XoEqT2Y4DJmf"
      },
      "source": [
        "### Import libraries and define constants"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "YYtGjGG45ELJ"
      },
      "source": [
        "Define constants used in this tutorial."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "5zmD19ryCre7"
      },
      "outputs": [],
      "source": [
        "PATH=%env PATH\n",
        "%env PATH={PATH}:/home/jupyter/.local/bin\n",
        "\n",
        "USER = \"your-user-name\"  # <---CHANGE THIS\n",
        "PIPELINE_ROOT = \"{}/pipeline_root/{}\".format(BUCKET_NAME, USER)\n",
        "\n",
        "PIPELINE_ROOT"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "IprQaSI25oSk"
      },
      "source": [
        "Import packages used in this tutorial."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "UFDUBveR5UfJ"
      },
      "outputs": [],
      "source": [
        "from typing import NamedTuple\n",
        "\n",
        "from kfp import dsl\n",
        "from kfp.v2 import compiler\n",
        "from kfp.v2.dsl import component\n",
        "from kfp.v2.google.client import AIPlatformClient"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "IbN_49SUW7b7"
      },
      "source": [
        "## Define a simple pipeline\n",
        "\n",
        "This example defines a simple pipeline that has three steps.\n",
        "The following sections define some pipeline *components*, and then defines a pipeline that uses them.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "nbV9dECIePV8"
      },
      "source": [
        "### Create Python function-based pipeline components\n",
        "\n",
        "First, create a component based on a very simple Python function. It takes a string input parameter and returns that value as output.\n",
        "\n",
        "Note the use of the `@component` decorator, which compiles the function to a KFP component when evaluated.  For example purposes, this example specifies a base image to use for the component (`python:3.9`), and a component YAML file, `hw.yaml`. The compiled component spec will be written to this file.  (The default base image is `python:3.7`, which would of course work just fine too).\n",
        "\n",
        "Next, run the cell below to generate the component yaml file in your local directory."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Ye7mr8WPW7b8"
      },
      "outputs": [],
      "source": [
        "@component(output_component_file=\"hw.yaml\", base_image=\"python:3.9\")\n",
        "def hello_world(text: str) -> str:\n",
        "    print(text)\n",
        "    return text"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "2dTdxgSm17Rk"
      },
      "source": [
        "As you'll see below, compilation of this component creates a [task factory function](https://www.kubeflow.org/docs/components/pipelines/sdk/python-function-components/)—called `hello_world`— that you can use in defining a pipeline step.  \n",
        "\n",
        "While not shown here, if you want to share this component definition, or use it in another context, you could also load it from its yaml file like this:\n",
        "`hello_world_op = components.load_component_from_file('./hw.yaml')`.  \n",
        "You can also use the `load_component_from_url` method, if your component yaml file is stored online. (For GitHub URLs, load the 'raw' file.)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "fkKFuQYDeFLB"
      },
      "source": [
        "Next, define two additional pipeline components.  \n",
        "\n",
        "For example purposes, the first component below, `two_outputs`, installs the given `google-cloud-storage` package. \n",
        "To keep the example code simple, the component function code won't actually use this import. \n",
        "\n",
        "This is one way that you can install the necessary package components.  \n",
        "Alternatively, you can specify a base image that includes the necessary installations.\n",
        "\n",
        "The `two_outputs` component returns two named outputs. In the next section, you'll see how those outputs can be consumed by other pipeline steps. \n",
        "\n",
        "The second component below, `consumer`, takes three string inputs and prints them out."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "HQNJpcK_m7YN"
      },
      "outputs": [],
      "source": [
        "@component(packages_to_install=[\"google-cloud-storage\"])\n",
        "def two_outputs(\n",
        "    text: str,\n",
        ") -> NamedTuple(\n",
        "    \"Outputs\",\n",
        "    [\n",
        "        (\"output_one\", str),  # Return parameters\n",
        "        (\"output_two\", str),\n",
        "    ],\n",
        "):\n",
        "    # the import is not actually used for this simple example, but the import\n",
        "    # is successful, as it was included in the `packages_to_install` list.\n",
        "    from google.cloud import storage  # noqa: F401\n",
        "\n",
        "    o1 = f\"output one from text: {text}\"\n",
        "    o2 = f\"output two from text: {text}\"\n",
        "    print(\"output one: {}; output_two: {}\".format(o1, o2))\n",
        "    return (o1, o2)\n",
        "\n",
        "\n",
        "@component\n",
        "def consumer(text1: str, text2: str, text3: str):\n",
        "    print(f\"text1: {text1}; text2: {text2}; text3: {text3}\")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "YTh_chI1PEpb"
      },
      "source": [
        "### Define a pipeline that uses the components\n",
        "\n",
        "Next, define a pipeline that uses these three components.\n",
        "\n",
        "By evaluating the component definitions above, you've created task factory functions that are used in the pipeline definition to create pipeline steps.\n",
        "\n",
        "The pipeline takes an input parameter, and passes that parameter as an argument to the first two pipeline steps (`hw_task` and `two_outputs_task`).  \n",
        "\n",
        "Then, the third pipeline step (`consumer_task`) consumes the outputs of the first and second steps.  Because the `hello_world` component definition just returns one unnamed output, you refer to it as `hw_task.output`.  The `two_outputs` task returns two named outputs, which you access as `two_outputs_task.outputs[\"<output_name>\"]`.\n",
        "\n",
        "Note: in the `@dsl.pipeline` decorator, you're defining the `PIPELINE_ROOT` Cloud Stora path to use.  If you had not included that info here, it would be required to specify it when creating the pipeline run, as you'll see below."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "UZ32XAM_PEpf"
      },
      "outputs": [],
      "source": [
        "@dsl.pipeline(\n",
        "    name=\"hello-world-v2\",\n",
        "    description=\"A simple intro pipeline\",\n",
        "    pipeline_root=PIPELINE_ROOT,\n",
        ")\n",
        "def intro_pipeline(text: str = \"hi there\"):\n",
        "    hw_task = hello_world(text)\n",
        "    two_outputs_task = two_outputs(text)\n",
        "    consumer_task = consumer(  # noqa: F841\n",
        "        hw_task.output,\n",
        "        two_outputs_task.outputs[\"output_one\"],\n",
        "        two_outputs_task.outputs[\"output_two\"],\n",
        "    )"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "2Hl1iYEKSzjP"
      },
      "source": [
        "## Compile and run the pipeline\n",
        "\n",
        "Next, you compile the pipeline.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "7PwUFV-MleGs"
      },
      "outputs": [],
      "source": [
        "from kfp.v2 import compiler  # noqa: F811\n",
        "\n",
        "compiler.Compiler().compile(\n",
        "    pipeline_func=intro_pipeline, package_path=\"hw_pipeline_job.json\"\n",
        ")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "qfNuzFswBB4g"
      },
      "source": [
        "The pipeline compilation generates the `hw_pipeline_job.json` job spec file.\n",
        "\n",
        "Next, instantiate an API client object:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Hl5Q74_gkW2c"
      },
      "outputs": [],
      "source": [
        "from kfp.v2.google.client import AIPlatformClient  # noqa: F811\n",
        "\n",
        "api_client = AIPlatformClient(\n",
        "    project_id=PROJECT_ID,\n",
        "    region=REGION,\n",
        ")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_jrn6saiQsPh"
      },
      "source": [
        "Then, you run the defined pipeline like this: "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "R4Ha4FoDQpkd"
      },
      "outputs": [],
      "source": [
        "response = api_client.create_run_from_job_spec(\n",
        "    job_spec_path=\"hw_pipeline_job.json\",\n",
        "    # pipeline_root=PIPELINE_ROOT  # this argument is necessary if you did not specify PIPELINE_ROOT as part of the pipeline definition.\n",
        ")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "GvBTCP318RKs"
      },
      "source": [
        "Click on the generated link to see your run in the Cloud Console.  It should look like this:\n",
        "\n",
        "<a href=\"https://storage.googleapis.com/amy-jo/images/mp/intro_pipeline.png\" target=\"_blank\"><img src=\"https://storage.googleapis.com/amy-jo/images/mp/intro_pipeline.png\" width=\"60%\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "3QdnVXcwpZAi"
      },
      "source": [
        "## Recurring pipeline runs: create a scheduled pipeline job\n",
        "\n",
        "This section shows how to create a **scheduled pipeline job**.  You do this using the pipeline you already defined.\n",
        "\n",
        "Under the hood, the scheduled jobs are supported by the Cloud Scheduler and a Cloud Functions function.  Check first that the APIs for both of these services are enabled. \n",
        "You will need to first enable the [enable the Cloud Scheduler API](http://console.cloud.google.com/apis/library/cloudscheduler.googleapis.com) and the [Cloud Functions and Cloud Build APIs](https://console.cloud.google.com/flows/enableapi?apiid=cloudfunctions,cloudbuild.googleapis.com) if you have not already done so.  \n",
        "Note:you need to [create an App Engine app for your project](https://cloud.google.com/scheduler/docs/quickstart) if one does not already exist.\n",
        "\n",
        "\n",
        "See the [Cloud Scheduler](https://cloud.google.com/scheduler/docs/configuring/cron-job-schedules) documentation for more on the cron syntax.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kjksVdC_0xLt"
      },
      "source": [
        "Create a scheduled pipeline job, passing as an arg the job spec file that you compiled above.  \n",
        "\n",
        "Note that you can pass a `parameter_values` dict that specifies the pipeline input parameters you want to use."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "RvUu12Rj6G8Z"
      },
      "outputs": [],
      "source": [
        "# adjust time zone and cron schedule as necessary\n",
        "response = api_client.create_schedule_from_job_spec(\n",
        "    job_spec_path=\"hw_pipeline_job.json\",\n",
        "    schedule=\"2 * * * *\",\n",
        "    time_zone=\"America/Los_Angeles\",  # change this as necessary\n",
        "    parameter_values={\"text\": \"Hello world!\"},\n",
        "    # pipeline_root=PIPELINE_ROOT  # this argument is necessary if you did not specify PIPELINE_ROOT as part of the pipeline definition.\n",
        ")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "6Dvbvc_ksuxE"
      },
      "source": [
        "Once the scheduled job is created, you can see it listed in the [Cloud Scheduler](https://console.cloud.google.com/cloudscheduler/) panel in the Console.\n",
        "\n",
        "<a href=\"https://storage.googleapis.com/amy-jo/images/kf-pls/pipelines_scheduler.png\" target=\"_blank\"><img src=\"https://storage.googleapis.com/amy-jo/images/kf-pls/pipelines_scheduler.png\" width=\"95%\"/></a>\n",
        "\n",
        "You can test the setup from the Cloud Scheduler panel by clicking 'RUN NOW'.\n",
        "\n",
        "> **Note**: The implementation is using a Cloud Functions function, which you can see listed in the [Cloud Functions](https://console.cloud.google.com/functions/list) panel in the console as `templated_http_request-v1`.  \n",
        "Don't delete this function, as it will prevent the Cloud Scheduler jobs from actually kicking off the pipeline run.  If you do delete it, create a new scheduled job in order to recreate the function.   \n",
        "\n",
        "When you're done experimenting, you probably want to **PAUSE** your scheduled job from the Cloud Scheduler panel, so that the recurrent jobs do not keep running."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "O4gCLMgYNlJb"
      },
      "source": [
        "## Specifying a service account to use for a pipeline run\n",
        "\n",
        "By default, the [service account](https://cloud.google.com/iam/docs/service-accounts) used for your pipeline run is your [default compute engine service account](https://cloud.google.com/compute/docs/access/service-accounts#default_service_account).  \n",
        "However, you might want to run pipelines with permissions to access different roles than those configured for your default SA (e.g. perhaps using a more restricted set of permissions).\n",
        "If you want to execute your pipeline using a different service account, this is straightforward to do.  You just need to give the new service account the correct permissions. "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ZIStAYOVRfOl"
      },
      "source": [
        "### Create a service account \n",
        "\n",
        "See the [documentation](https://cloud.google.com/vertex-ai/docs/pipelines/configure-project#service-account) for information on the process of creating and configuring a service account to work with Vertex Pipelines. \n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XrH_JBGETX4g"
      },
      "source": [
        "Once your service account is configured, you can pass it as an arg to the `create_run_from_job_spec` method, as follows.  Set `SERVICE_ACCOUNT` to your info and uncomment the `service_account` arg in `create_run_from_job_spec`:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "V_0rTpLNAEt2"
      },
      "outputs": [],
      "source": [
        "SERVICE_ACCOUNT = (\n",
        "    \"<service-account-name>@your-project-id.iam.gserviceaccount.com\"  # <--- CHANGE THIS\n",
        ")\n",
        "\n",
        "api_client = AIPlatformClient(\n",
        "    project_id=PROJECT_ID,\n",
        "    region=REGION,\n",
        ")\n",
        "\n",
        "response = api_client.create_run_from_job_spec(\n",
        "    job_spec_path=\"hw_pipeline_job.json\",  # <--- CHANGE THIS IF YOU WANT TO RUN OTHER PIPELINES\n",
        "    pipeline_root=PIPELINE_ROOT,\n",
        "    #     service_account=SERVICE_ACCOUNT,  # <-- UNCOMMENT to use non-default service account\n",
        ")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ZjweeCVdgB5w"
      },
      "source": [
        "The pipeline job runs with the permissions of the given service account."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "c2835a82e338"
      },
      "source": [
        "## Pipeline step caching\n",
        "\n",
        "By default, pipeline step caching is enabled. This means that the results of previous step executions are reused when possible. \n",
        "\n",
        "if you want to disable caching for a pipeline run, you can pass the `enable_caching=False` arg to the `create_run_from_job_spec` function when you submit the pipeline job, as shown below.  Try submitting the example pipeline job again, first with and then without caching enabled."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "a08eac7d77a3"
      },
      "outputs": [],
      "source": [
        "# response = api_client.create_run_from_job_spec(\n",
        "#     job_spec_path=\"hw_pipeline_job.json\",\n",
        "#     enable_caching=False\n",
        "# )"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Xm46quDiBp7M"
      },
      "source": [
        "## Using the Pipelines REST API\n",
        "\n",
        "At times you may want to use the REST API instead of the Python KFP SDK.  Below are examples of how to do that.\n",
        "\n",
        "Where a command requires a pipeline ID, you can get that info from the \"Run\" column in the pipelines list— as shown below— as well as from the 'details' page for a given pipeline.  You can also see that info when you list pipeline job(s) via the API.\n",
        "\n",
        "<a href=\"https://storage.googleapis.com/amy-jo/images/mp/pipeline_id.png\" target=\"_blank\"><img src=\"https://storage.googleapis.com/amy-jo/images/mp/pipeline_id.png\" width=\"80%\"/></a>\n",
        "\n",
        "> Note: Currently, the KFP SDK does not support all the API calls below.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "sby6SIGXB_3O"
      },
      "outputs": [],
      "source": [
        "ENDPOINT = REGION + \"-aiplatform.googleapis.com\""
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "oPPjZYXnCa_W"
      },
      "source": [
        "### List pipeline jobs\n",
        "\n",
        "(This request may generate a large response if you have many pipeline runs)."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "hAC_ZCMZJKTJ"
      },
      "outputs": [],
      "source": [
        "! curl -X GET -H \"Authorization: Bearer $(gcloud auth print-access-token)\"   -H \"Content-Type: application/json\"   https://{ENDPOINT}/v1beta1/projects/{PROJECT_ID}/locations/{REGION}/pipelineJobs"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Je-HJUUp0MAi"
      },
      "source": [
        "### Create a pipeline job\n",
        "\n",
        "For this API request, you need to submit a compiled pipeline job spec.  We're using the one we generated above, \"hw_pipeline_job.json\"."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "fXDF-br4r220"
      },
      "outputs": [],
      "source": [
        "! curl -X POST  -H \"Authorization: Bearer $(gcloud auth print-access-token)\" -H \"Content-Type: application/json; charset=utf-8\"   https://{ENDPOINT}/v1beta1/projects/{PROJECT_ID}/locations/{REGION}/pipelineJobs  --data \"@sample_pipeline.json\""
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "cjuuG8GRCiNe"
      },
      "source": [
        "### Get a pipeline job from its ID"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "BFtoTo64gnqV"
      },
      "outputs": [],
      "source": [
        "PIPELINE_RUN_ID = \"xxxx\"  # <---CHANGE THIS\n",
        "! curl -X GET -H \"Authorization: Bearer $(gcloud auth print-access-token)\"   -H \"Content-Type: application/json\"   https://{ENDPOINT}/v1beta1/projects/{PROJECT_ID}/locations/{REGION}/pipelineJobs/{PIPELINE_RUN_ID}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "RDKhnFrNCol3"
      },
      "source": [
        "### Cancel a pipeline job given its ID"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "6efo3RjWhFDp"
      },
      "outputs": [],
      "source": [
        "PIPELINE_RUN_ID_TO_CANCEL = \"xxxx\"  # <---CHANGE THIS\n",
        "! curl -X POST -H \"Authorization: Bearer $(gcloud auth print-access-token)\"   -H \"Content-Type: application/json\"   https://{ENDPOINT}/v1beta1/projects/{PROJECT_ID}/locations/{REGION}/pipelineJobs/{PIPELINE_RUN_ID_TO_CANCEL}:cancel"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "s9Tc0cO8C1OF"
      },
      "source": [
        "### Delete a pipeline job given its ID"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "lu2IAHhWijDY"
      },
      "outputs": [],
      "source": [
        "PIPELINE_RUN_ID_TO_DELETE = \"xxxx\"  # <---CHANGE THIS\n",
        "! curl -X DELETE -H \"Authorization: Bearer $(gcloud auth print-access-token)\"   -H \"Content-Type: application/json\"   https://{ENDPOINT}/v1beta1/projects/{PROJECT_ID}/locations/{REGION}/pipelineJobs/{PIPELINE_RUN_ID_TO_DELETE}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "6yRt6JlmrGuG"
      },
      "source": [
        "## Cleaning up\n",
        "\n",
        "To clean up all Google Cloud resources used in this project, you can [delete the Google Cloud\n",
        "project](https://cloud.google.com/resource-manager/docs/creating-managing-projects#shutting_down_projects) you used for the tutorial.\n",
        "\n",
        "Otherwise, you can delete the individual resources you created in this tutorial:\n",
        "- delete Cloud Storage objects that were created.  Uncomment and run the command in the cell below **only if you are not using the `PIPELINE_ROOT` path for any other purpose**.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "mkdPFlpwrGuP"
      },
      "outputs": [],
      "source": [
        "# Warning: this command will delete ALL Cloud Storage objects under the PIPELINE_ROOT path.\n",
        "# ! gsutil -m rm -r $PIPELINE_ROOT"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "collapsed_sections": [],
      "name": "pipelines_intro_kfp.ipynb",
      "toc_visible": true
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}
